package com.udf.flink.scala.apitest.table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row
/**
 * Author z
 * Date 2021-02-26 11:21:36
 */
object Temporaljoin {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val bsSettings = EnvironmentSettings.newInstance().inStreamingMode().build()
    val tableEnv = StreamTableEnvironment.create(env, bsSettings)

    // 获取左侧流
    val orderStream = env
      .socketTextStream("localhost", 9998)
      .map {
        x =>
          val s = x.split(",")
          (s(0), s(1).toInt)  // sh,1
      }
    // 获取右侧 时态表流
    val cityStream = env
      .socketTextStream("localhost", 9999)
      .map {
        x =>
          val s = x.split(",")
          (s(0).toInt, s(1))
      }
    // 将流转换为table，并设置p 为处理时间属性

    val orderTable = tableEnv.fromDataStream(orderStream,Schema
      .newBuilder()
      .column("_1", "string")
      .column("_2", "int")
      .columnByExpression("p", "PROCTIME()") // .watermark("f1", "`f1` - INTERVAL '5' SECOND")
      .build()
     ).as("oder_name", "city_id", "p")
     // 老式写法 .fromDataStream(orderStream, 'oder_name, 'city_id, 'p.proctime())
    //.fromDataStream(orderStream,$"oder_name",$"city_id",$"p".proctime())
    // 将流转换为table，并设置p 为处理时间属性

    val cityTable = tableEnv
      .fromDataStream(cityStream, 'city_id, 'city_name, 'p.proctime())
    tableEnv.toDataStream(orderTable).print()
    tableEnv.toRetractStream[Row](cityTable).print()
    tableEnv.createTemporaryView("cityview", cityTable)
    tableEnv.createTemporaryView("orderview", orderTable)

    // 创建时态表 function
    val cityFunc = cityTable
      .createTemporalTableFunction('p, 'city_id)
    tableEnv.createTemporaryFunction("cityFunc", cityFunc)

    // 执行join
    val result = tableEnv.sqlQuery(
      """
        |select
        |a.oder_name,
        |a.city_id,
        |b.city_name
        |from orderview a,
        |lateral table (cityFunc(a.p)) b
        |where a.city_id = b.city_id
        |""".stripMargin)
    tableEnv.toRetractStream[Row](result).print()
    env.execute()
  }
}
